package cn.parus.service.Impl;

import cn.parus.common.utils.UUidUtils;
import cn.parus.service.HdfsService;
import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.io.InputStream;

/**
 * @Author: Axel
 * @Description: TODO
 **/
@Slf4j
@Service
public class HdfsServiceImpl implements HdfsService {

    @Autowired
    private FileSystem hdfsFileSystem;

    @Value("${file.baseUrl}")
    private String baseUrl;

    /**
     * 从服务器端上传到hdfs
     * @param localFilePath
     * @return
     * @throws IOException
     */
    public String uploadFileToHDFS(String localFilePath) throws IOException {
        // 构造文件原路径
        Path localPath = new Path(System.getProperty("user.dir") + baseUrl + localFilePath);
        // 拼接文件的唯一id
        String UUID =  UUidUtils.generateShortUUid();
        Path hdfsPath = new Path("/" + UUID);
        hdfsFileSystem.copyFromLocalFile(localPath, hdfsPath);
        /**
         * hdfsFileSystem.setOwner(hdfsPath, Owner, null);
         * 已经禁用了用户验证，直接使用即可
         */
        return UUID;
    }

    /**
     * 从hdfs中下载到服务器端
     * @param hdfsFilePath
     * @throws IOException
     */
    public String downloadFileFromHDFS(String hdfsFilePath) throws IOException {
        Path hdfsPath = new Path(hdfsFilePath);
        String localFilePath = System.getProperty("user.dir") + baseUrl;
        Path localPath = new Path(localFilePath);
        hdfsFileSystem.copyToLocalFile(hdfsPath, localPath);
        return localFilePath;
    }

    public void deleteFileOrDirInHDFS(String hdfsFilePath) throws IOException {
        Path path = new Path(hdfsFilePath);
        hdfsFileSystem.delete(path, true); // 递归删除
    }

    public void createDirInHDFS(String hdfsDirPath) throws IOException {
        Path path = new Path(hdfsDirPath);
        hdfsFileSystem.mkdirs(path);
    }
}
